﻿/*
   Copyright 2012 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Security.Principal;
using System.Text;
using Flower.Actors;
using Flower.Client;
using Flower.Logging;
using System.Threading;
using Flower.Directory.Util;
using Flower.Processing;
using Flower.Services;
using Flower.Workflow;
using Flower.Processing.Interruptions;
using Flower.Security;
using System.Runtime.Remoting.Lifetime;
using System.Threading.Tasks;

namespace Flower.Processor
{
    internal class ProcessorHost : MarshalByRefObject, IProcessorHost
    {
        private abstract class Command : IMessage<ProcessorHost>
        {
            public abstract void Process(ProcessorHost processor);
        }

        private class LoadAndExecuteProcessCommand : Command
        {
            public string Pid;

            public override void Process(ProcessorHost processor)
            {
                processor.Process(this);
            }
        }

        private class ExecuteActivityCommand : Command
        {
            public Processing.Processor Processor;

            public override void Process(ProcessorHost processor)
            {
                processor.Process(this);
            }
        }

        private static readonly Log Log = LogManager.CreateLog("Processor:" + AppDomain.CurrentDomain.FriendlyName);

        private static readonly TimeSpan RolesRefreshInterval = TimeSpan.FromMinutes(1);

        private readonly ProcessorSettings _settings = ApplicationContext.GetProcessorSettings();
        private readonly IFlowerClient _flowerClient = ApplicationContext.GetDirectoryClient();
        private readonly IStateService _stateService = ApplicationContext.GetStateService();
        private readonly ClientSponsor _sponsor = new ClientSponsor();

        private readonly IAuthorizationProvider _authorizationProvider = 
            new AuthorizationProvider(RolesRefreshInterval) 
            {
                Directory = ApplicationContext.GetDirectoryClient().Directory
            };

        private IActor<Command> _loader;
        private IActor<Command> _blockingProcessor;
        private IActor<Command> _nonBlockingProcessor;

        private readonly ObjectPool<LoadAndExecuteProcessCommand> _loadAndExecuteProcessCommandPool =
            new ObjectPool<LoadAndExecuteProcessCommand>(null, cmd => cmd.Pid = null);

        private ProcessorsCache _processorsCache;
        private IProcessingManager _processingManager;

        private volatile bool _isStopped;

        private readonly ConcurrentDictionary<string, ManualResetEventSlim> _pidLocks =
            new ConcurrentDictionary<string, ManualResetEventSlim>();
        private readonly CountdownEvent _asyncTasksCount = new CountdownEvent(0);

        public AppDomain AppDomain
        {
            get { return AppDomain.CurrentDomain; }
        }

        #region Utils

        private IActor<Command> SelectActorToExecuteNextActivity(Processing.Processor processor)
        {
            return (processor.NextActivity.IsBlocking) ? _blockingProcessor : _nonBlockingProcessor;
        }

        private bool IsAllowedToExecute(Processing.Processor processor)
        {
            var attributes = 
                processor
                    .State
                    .GetType()
                    .GetCustomAttributes(typeof(AllowExecuteToAttribute), false);

            if (attributes.Length == 0)
            {
                return true;
            }

            string[] roles = _authorizationProvider.GetUserRoles(WindowsIdentity.GetCurrent());

            foreach (var attr in attributes.Select(attr => (AllowExecuteToAttribute)attr))
            {
                if (roles.Any(r => PermissionCheck.RolesMatch(r, attr.Role)))
                {
                    return true;
                }
            }

            return false;
        }

        #endregion

        #region Start/stop

        public void Start(IProcessingManager processingManager)
        {
            if (_isStopped) throw new InvalidOperationException("The processor is stopped.");
            if (processingManager == null) throw new ArgumentNullException("processingManager");

            _processingManager = processingManager;

            var procManagerMarshalByRef = processingManager as MarshalByRefObject;
            if (procManagerMarshalByRef != null)
            {
                _sponsor.Register(procManagerMarshalByRef);
            }

            Log.Trace(() => string.Format("Starting processor host {0}.", AppDomain.CurrentDomain.FriendlyName));

            _processorsCache = new ProcessorsCache();

            _loader =
                new MultiActor<ProcessorHost, Command>
                (
                    _settings.LoadingThreadsCount,
                    i => new Actor<ProcessorHost, Command>(this)
                );
            _blockingProcessor =
                new MultiActor<ProcessorHost, Command>
                (
                    _settings.BlockingThreadsCount,
                    i => new Actor<ProcessorHost, Command>(this)
                );
            _nonBlockingProcessor =
                new MultiActor<ProcessorHost, Command>
                (
                    _settings.NonBlockingThreadsCount,
                    i =>
                        new Actor<ProcessorHost, Command>
                        (
                            this,
                            new ConcurrentQueue<Command>(),
                            0,
                            _settings.NonBlockingActivityTimeout
                        )
                );

            _blockingProcessor.Error += (sender, args) => 
                Log.Error
                (
                    string.Format
                    (
                        "Exception occured while processing the command {0}.", 
                        args.Message.GetType().Name
                    ), 
                    args.Exception
                );
            _nonBlockingProcessor.Error += (sender, args) =>
                Log.Error
                (
                    string.Format
                    (
                        "Exception occured while processing the command {0}.",
                        args.Message.GetType().Name
                    ),
                    args.Exception
                );

            Log.Trace(() => string.Format("Processor host {0} is started.", AppDomain.CurrentDomain.FriendlyName));
        }

        public void Stop()
        {
            Log.Trace(() => string.Format("Stopping processor host {0}.", AppDomain.CurrentDomain.FriendlyName));

            foreach (ManualResetEventSlim pidLock in _pidLocks.Values)
            {
                pidLock.Set();
            }

            Parallel.ForEach(new[] { _loader, _blockingProcessor, _nonBlockingProcessor }, a => a.Stop());

            _blockingProcessor.Dispose();
            _nonBlockingProcessor.Dispose();
            _processorsCache.Dispose();

            _asyncTasksCount.Wait();
            _asyncTasksCount.Dispose();

            var procManagerMarshalByRef = _processingManager as MarshalByRefObject;
            if (procManagerMarshalByRef != null)
            {
                _sponsor.Unregister(procManagerMarshalByRef);
            }

            _isStopped = true;

            _flowerClient.Dispose();

            Log.Trace(() => string.Format("Processor host {0} is stopped.", AppDomain.CurrentDomain.FriendlyName));
        }

        #endregion

        #region Command handlers

        private void Process(LoadAndExecuteProcessCommand command)
        {
            Log.Trace(() => string.Format("Loading process {0}.", command.Pid));

            try
            {
                ManualResetEventSlim pidLock;
                if (_pidLocks.TryGetValue(command.Pid, out pidLock))
                {
                    pidLock.Wait();
                }

                _processingManager.MarkRunning(command.Pid);

                Processing.Processor processor;

                try
                {
                    processor = _processorsCache.Take(command.Pid);

                    if (processor == null)
                    {
                        processor =
                            new Processing.Processor
                            (
                                _settings.ProcessorName,
                                _flowerClient,
                                _stateService,
                                new Processing.Processor.Settings
                                {
                                    ProcessorTransactionTimeout = _settings.ProcessorTransactionTimeout,
                                    MinWaitOnErrorSeconds = _settings.MinWaitOnErrorSeconds,
                                    MaxRetries = _settings.MaxRetries
                                }
                            );
                        processor.LoadProcess(command.Pid, _settings.CustomAdvisors);

                        if (!IsAllowedToExecute(processor))
                        {
                            Log.Error
                            (
                                string.Format("This processor is not allowed to execute processes of the workflow type {0}.", processor.GetType().AssemblyQualifiedName)
                            );
                            _processingManager.MarkBroken(command.Pid);
                            return;
                        }
                    }

                    if (processor.IsFinished)
                    {
                        Log.Trace(string.Format("Cannot execute the process {0} because it is finished.", command.Pid));
                        _processingManager.MarkFinished(command.Pid);
                        return;
                    }
                }
                catch (Exception ex)
                {
                    _processingManager.MarkBroken(command.Pid);
                    Log.Error(string.Format("Failed to load process {0}.", command.Pid), ex);
                    return;
                }

                try
                {
                    SelectActorToExecuteNextActivity(processor)
                        .Send(new ExecuteActivityCommand { Processor = processor });
                }
                catch (Exception ex)
                {
                    _processingManager.MarkBroken(command.Pid);
                    Log.Error(string.Format("Failed to load process {0}.", command.Pid), ex);
                    return;
                }
            }
            finally
            {
                _loadAndExecuteProcessCommandPool.Return(command);
            }
        }

        private void Process(ExecuteActivityCommand command)
        {
            Interruption interruption;
            ManualResetEventSlim pidLock;

            _pidLocks.TryAdd(command.Processor.Pid, new ManualResetEventSlim(false));

            try
            {
                interruption = command.Processor.ExecuteActivity();
            }
            catch (ThreadAbortException)
            {
                Thread.ResetAbort();
                _processingManager.MarkBroken(command.Processor.Pid);

                if (_pidLocks.TryRemove(command.Processor.Pid, out pidLock))
                {
                    pidLock.Set();
                }
                return;
            }

            if (interruption == null || interruption is Breakpoint)
            {
                SelectActorToExecuteNextActivity(command.Processor).Send(command);

                if (_pidLocks.TryRemove(command.Processor.Pid, out pidLock))
                {
                    pidLock.Set();
                }
                return;
            }

            var asyncTask = interruption as Async;
            if (asyncTask != null)
            {
                _asyncTasksCount.AddCount();
                asyncTask.Task.ContinueWith
                (
                    t =>
                    {
                        SelectActorToExecuteNextActivity(command.Processor).Send(command);
                        _asyncTasksCount.Signal();
                    }
                );

                if (_pidLocks.TryRemove(command.Processor.Pid, out pidLock))
                {
                    pidLock.Set();
                }
                return;
            }

            var waiting = interruption as Waiting;
            if (waiting != null)
            {
                _processingManager.MarkWaiting(command.Processor.Pid);
                _processorsCache.Put(command.Processor);

                if (_pidLocks.TryRemove(command.Processor.Pid, out pidLock))
                {
                    pidLock.Set();
                }
                return;
            }

            var suspention = interruption as Suspention;
            if (suspention != null)
            {
                _processingManager.MarkSuspended(command.Processor.Pid, suspention.Until, suspention.LinkId);
                _processorsCache.Put(command.Processor);

                if (_pidLocks.TryRemove(command.Processor.Pid, out pidLock))
                {
                    pidLock.Set();
                }
                return;
            }

            var unhandledException = interruption as UnhandledException;
            if (unhandledException != null)
            {
                _processingManager.MarkBroken(command.Processor.Pid);

                if (_pidLocks.TryRemove(command.Processor.Pid, out pidLock))
                {
                    pidLock.Set();
                }
                return;
            }

            var finish = interruption as Finish;
            if (finish != null)
            {
                _processingManager.MarkFinished(command.Processor.Pid);

                if (_pidLocks.TryRemove(command.Processor.Pid, out pidLock))
                {
                    pidLock.Set();
                }
                return;
            }

            throw new ArgumentOutOfRangeException(string.Format("Unknown interruption type {0}.", interruption.GetType().FullName));
        }

        #endregion

        #region IProcessorHost implementation

        public void LoadAndExecuteProcess(string pid)
        {
            if (_isStopped) throw new InvalidOperationException("The processor is stopped.");

            LoadAndExecuteProcessCommand cmd = _loadAndExecuteProcessCommandPool.Get();
            cmd.Pid = pid;
            _loader.Send(cmd);
        }

        public void ResumeActors()
        {
            Log.Trace("Resuming all actors.");

            _loader.Resume();
            _blockingProcessor.Resume();
        }

        #endregion
    }
}
